#define __THREADPOOL_C__
#include <unistd.h>
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <assert.h>
#include <sys/types.h>   
#include <string.h>

#include "Threadpool.h"

void* thread_function(void* arg)
{
  struct threadpool* pool =(struct threadpool*)arg;
  struct task* ptask =NULL;
  while(1)
 {
    pthread_mutex_lock(&pool->mutex);
    while((pool->quene_cur_num==0) &&(!pool->pool_close))
    {
       //任务队列为空的时候，等待队列非空
        pthread_cond_wait(&pool->quene_not_empty,&pool->mutex);
    }
    if(pool->pool_close) //如果线程池关闭，就退出线程
    {
       pthread_mutex_unlock(&pool->mutex);
       pthread_exit(NULL);
    }
    pool->quene_cur_num--;
    ptask=pool->head;
    if(pool->quene_cur_num==0)
    {
       pool->head=pool->tail=NULL;
    }
    else
   {
      pool->head=ptask->next;
   }
   if(pool->quene_cur_num==0)
    {
      //任务队列为空，通知销毁，
     pthread_cond_broadcast(&pool->quene_empty);
         //pthread_cond_signal(&pool->quene_empty);
    } 
   if(pool->quene_cur_num == pool->quene_max_num-1)
   {
     //任务队列没有满，通知threadpool_add_task 添加新的任务
      pthread_cond_broadcast(&pool->quene_not_full);
   }
   pthread_mutex_unlock(&pool->mutex);
   (*(ptask->callback_function))(ptask->arg);
   free(ptask);
   ptask=NULL;
 } 
  return NULL;
}
struct threadpool* thread_pool_init(int thread_num,int quene_max_num)
{
   int i;
   struct threadpool* pool =NULL;
   pool=(struct threadpool*)malloc(sizeof(struct threadpool));
   if(pool==NULL)
    {
        printf("malloc pool failed!\n");
        return NULL;
    }
   pool->thread_num = thread_num;
   pool->quene_max_num=quene_max_num;
   pool->head =NULL;
   pool->tail =NULL;
   pool->quene_cur_num =0;
   pool->threadids =(pthread_t *)malloc(sizeof(pthread_t) * thread_num);
   if(pool->threadids==NULL)
    {
        printf("malloc threadids failed!\n");
        return NULL;
    }
   if(pthread_mutex_init(&(pool->mutex),NULL))
    {
        printf("init mutex failed!\n");
        return NULL;       
    }
   if(pthread_cond_init(&(pool->quene_empty),NULL))
    {
        printf("init quene_empty failed!\n");
        return NULL;       
    }
   if(pthread_cond_init(&(pool->quene_not_empty),NULL))
    {
        printf("init quene_not_empty failed!\n");
        return NULL;       
    }
   if(pthread_cond_init(&(pool->quene_not_full),NULL))
    {
        printf("init quene_not_full failed!\n");
        return NULL;       
    }
   pool->quene_close =0;
   pool->pool_close =0;
   for(i=0;i<thread_num;i++)
   {
      pthread_create(&pool->threadids[i],NULL,thread_function,(void*)pool);
   }
   return pool;
}
int threadpool_add_task(struct threadpool* pool,CALLBACK callback_function,void* arg)
{
   assert(pool!=NULL);
   assert(callback_function!=NULL);
   assert(arg!=NULL);
   
   pthread_mutex_lock(&pool->mutex);
  while((pool->quene_cur_num ==pool->quene_max_num)&&(!pool->quene_close ||!pool->pool_close))
   {
     //任务队列满的时候，等待
     pthread_cond_wait(&pool->quene_not_full,&pool->mutex);
   }
   if(pool->quene_close ||pool->pool_close)
   {
       pthread_mutex_unlock(&pool->mutex);
       return -1;
   }
  struct task* ptask=(struct task*)malloc(sizeof(struct task));
  if(NULL==ptask)
  {
      pthread_mutex_unlock(&pool->mutex);
       return -1;   
  }
  ptask->callback_function =callback_function;
  ptask->arg =arg;
  ptask->next =NULL;

  if(NULL==pool->head)
  {
     pool->head=pool->tail =ptask;
     pthread_cond_broadcast(&pool->quene_not_empty);
  }  
  else
  {
    pool->tail->next=ptask;
    pool->tail =ptask;
  }
  pool->quene_cur_num++;
  pthread_mutex_unlock(&pool->mutex);
  return 0;
}

int threadpool_destroy(struct threadpool* pool)
{
  assert(pool!=NULL);
  pthread_mutex_lock(&pool->mutex);
  if(pool->quene_close ||pool->quene_close)
  {
      pthread_mutex_unlock(&pool->mutex);
      return -1;
  }
  pool->quene_close=1;
   while(pool->quene_cur_num !=0)
  {
     pthread_cond_wait(&pool->quene_empty,&pool->mutex);
  }
  pool->pool_close =1;
  pthread_mutex_unlock(&pool->mutex);
  //唤醒目前线程池中的阻塞线程
  pthread_cond_broadcast(&pool->quene_not_empty);
 //唤醒添加任务的threadpool_add_task
  pthread_cond_broadcast(&pool->quene_not_full);
  
  int i;
  //for(i=0;i<pool->quene_cur_num;i++) // modify
 for(i=0;i<pool->thread_num;i++)
  {
     pthread_join(pool->threadids[i],NULL);
  }
  pthread_mutex_destroy(&pool->mutex);
  pthread_cond_destroy(&pool->quene_empty);
  pthread_cond_destroy(&pool->quene_not_empty);
  pthread_cond_destroy(&pool->quene_not_full);
  
  free(pool->threadids);
  struct task* p;
  while(pool->head!=NULL)
  {
     p=pool->head;
     pool->head=p->next;
     free(p);
  }
  free(pool);
  return 0;
}
